Harden bridge transport and compatibility - #18

tests/test_security_hardening.py (3)

337-340: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Bound the disconnect() await so a regression fails instead of hanging.

await first.disconnect() has no time limit. This test depends on disconnect() cancelling the running _sync_worker task. If that cancellation regresses, the test blocks until the suite-level timeout rather than reporting a failure.

🧪 Proposed change

-        await first.disconnect()
+        await asyncio.wait_for(first.disconnect(), timeout=1.0)
         claimed = await second._claim_pending_users()

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_security_hardening.py` around lines 337 - 340, Bound the await of
first.disconnect() in the test around _claim_pending_users so cancellation
regressions fail promptly instead of hanging; use the test suite’s existing
timeout utility or convention and preserve the subsequent claimed-user
assertions.

---

158-183: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Consider extracting the hand-built worker fixture.

This test assigns about eighteen attributes to a GrpcNode created with __new__. SharedStoreDisconnectTests._controller performs a similar setup. When _sync_worker starts reading a new attribute, these tests fail with AttributeError instead of a meaningful assertion, and each fixture must be updated separately.

Extract a shared module-level builder that both test classes call.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_security_hardening.py` around lines 158 - 183, Extract the
repeated hand-built GrpcNode setup from
test_stream_open_timeout_increments_worker_failure_and_requeues and
SharedStoreDisconnectTests._controller into a shared module-level builder. Have
both tests call the builder, while preserving their scenario-specific overrides
and mocks, so newly required _sync_worker attributes are initialized in one
place.

---

261-281: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Move this test out of LoggingSafetyTests.

test_connect_restarts_worker_to_discover_stored_pending_work verifies worker restart behavior on connect. It does not verify logging safety. Place it in a class that describes worker lifecycle so the suite stays navigable.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_security_hardening.py` around lines 261 - 281, Move
test_connect_restarts_worker_to_discover_stored_pending_work out of
LoggingSafetyTests and into the existing test class covering worker lifecycle or
connect behavior. Keep the test setup, assertions, and mocking unchanged; only
relocate it to the semantically appropriate class.

tests/test_storage.py (1)

73-82: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Consider covering the claimed-user accounting and the constructor validation.

The new test covers pending-only accounting. Two behaviors added in this PR remain untested: enqueue_users counts claimed users toward the bound, and the constructor rejects a non-positive max_pending_users_per_node. Both are cheap to add.

🧪 Suggested additional tests

    async def test_claimed_users_count_toward_bound(self):
        store = InMemoryUserSyncStore(max_pending_users_per_node=1)
        await store.enqueue_users("node-1", [User(email="a@example.com")])
        await store.claim_users("node-1", "worker-1", limit=10, lease_seconds=30)

        with self.assertRaises(UserSyncStoreFullError):
            await store.enqueue_users("node-1", [User(email="b@example.com")])

    def test_non_positive_bound_is_rejected(self):
        with self.assertRaises(ValueError):
            InMemoryUserSyncStore(max_pending_users_per_node=0)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_storage.py` around lines 73 - 82, Add tests covering the remaining
constructor and accounting behavior in the storage test suite: add an async test
that claims the node’s only user and verifies enqueue_users rejects another user
because claimed users count toward max_pending_users_per_node, and add a
constructor test verifying InMemoryUserSyncStore rejects a zero or otherwise
non-positive bound with ValueError.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

PasarGuardNodeBridge/controller.py (1)

748-753: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Use the !s conversion flag for the exception text.

Ruff reports RUF010 on Line 751. The surrounding code already uses {e!s} in other log statements, for example in _cleanup_sync_worker at Line 532.

♻️ Proposed fix

-                f"[{self.name}] Unexpected error in sync worker | Error: {error_type} - {str(e)}", exc_info=True
+                f"[{self.name}] Unexpected error in sync worker | Error: {error_type} - {e!s}", exc_info=True

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@PasarGuardNodeBridge/controller.py` around lines 748 - 753, Update the
unexpected-error log in the sync worker’s exception handler to use the `!s`
conversion flag when formatting the exception text, while preserving the
existing error type, message context, and `exc_info=True` behavior.

Source: Linters/SAST tools

tests/test_security_hardening.py (1)

526-574: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Reduce timing sensitivity in the lease-expiry test.

The test depends on wall-clock margins that are small. first._sync_lease_seconds is 0.08, and Line 561 sleeps 0.02 before asserting that the second worker has not processed the claim. On a loaded CI runner, the second worker can claim the expired lease before that assertion runs, which makes the test flaky.

Increase the lease duration and the observation window so the margin between "lease still held" and "lease expired" is larger.

♻️ Proposed timing adjustment

-        first._sync_lease_seconds = 0.08
+        first._sync_lease_seconds = 0.5
...
-        await asyncio.sleep(0.02)
+        await asyncio.sleep(0.1)
         self.assertFalse(second_processed.is_set())
         self.assertFalse(second_worker.done())
 
-        await asyncio.wait_for(second_processed.wait(), timeout=0.5)
+        await asyncio.wait_for(second_processed.wait(), timeout=2.0)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_security_hardening.py` around lines 526 - 574, Adjust the timing
constants in test_second_worker_wakes_after_failed_requeue_lease_expires to
increase the lease duration and lengthen the pre-expiry observation delay,
preserving the assertion that the second worker has not processed the claim
before expiration and the existing post-expiry wait behavior.

tests/test_storage.py (1)

119-122: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Cover the negative boundary too.

The test name says “non-positive,” but it only checks 0. Add -1 so regressions in the validation condition are detected.

Suggested test adjustment

     def test_non_positive_per_node_bound_is_rejected(self):
-        with self.assertRaises(ValueError):
-            InMemoryUserSyncStore(max_pending_users_per_node=0)
+        for limit in (0, -1):
+            with self.subTest(limit=limit):
+                with self.assertRaises(ValueError):
+                    InMemoryUserSyncStore(max_pending_users_per_node=limit)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_storage.py` around lines 119 - 122, Update
test_non_positive_per_node_bound_is_rejected to also construct
InMemoryUserSyncStore with max_pending_users_per_node=-1 inside the ValueError
assertion, covering both zero and negative non-positive bounds.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

tests/test_user_revocation.py (1)

20-43: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Share the hand-built controller factory between test modules.

This _controller helper builds a Controller with object.__new__ and sets 21 private attributes by hand. tests/test_security_hardening.py defines a near-identical helper in SharedStoreDisconnectTests._controller. The two copies already differ: this one omits _tasks, _task_lock, and _version_lock.

When Controller.__init__ or _sync_worker starts using a new attribute, both copies must be updated, and a missed update surfaces as an AttributeError inside the worker rather than a clear failure. Move the factory into a shared test helper module.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_user_revocation.py` around lines 20 - 43, Move the hand-built
Controller factory from this test module into a shared test helper, then update
both this module and SharedStoreDisconnectTests._controller to import and reuse
it. Preserve the existing setup while consolidating all required private
attributes, including _tasks, _task_lock, and _version_lock, so future
Controller changes require updates in only one factory.

tests/test_security_hardening.py (1)

437-443: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

_wait_until spins the event loop instead of yielding time.

await asyncio.sleep(0) yields control but schedules an immediate callback. The loop therefore runs at full CPU for up to timeout. test_idle_retirement_boundary_100x_never_strands_enqueued_work calls this helper 100 times, so the cost accumulates.

Use a small positive sleep so the loop can idle between checks.

♻️ Proposed change

     `@staticmethod`
     async def _wait_until(predicate, timeout=0.2):
         async def poll():
             while not predicate():
-                await asyncio.sleep(0)
+                await asyncio.sleep(0.001)
 
         await asyncio.wait_for(poll(), timeout=timeout)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_security_hardening.py` around lines 437 - 443, Update the
_wait_until helper’s poll loop to await a small positive sleep interval instead
of asyncio.sleep(0), allowing the event loop to idle between predicate checks
while preserving the existing timeout and polling behavior.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

tests/test_security_hardening.py (1)

686-691: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Add an explicit strict= argument to zip().

Ruff flags B905 here. The two iterables differ in length by design, so strict=True would raise. Pass strict=False to make the intent explicit and clear the lint warning.

♻️ Proposed change

-            self.assertTrue(all(b > a for a, b in zip(store.claim_times, store.claim_times[1:])))
+            self.assertTrue(
+                all(b > a for a, b in zip(store.claim_times, store.claim_times[1:], strict=False))
+            )

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_security_hardening.py` around lines 686 - 691, Update the zip()
call in the claim_times ordering assertion to pass strict=False explicitly,
preserving the intentional behavior for iterables of differing lengths and
clearing Ruff B905.

Source: Linters/SAST tools

PasarGuardNodeBridge/abstract_node.py (1)

70-77: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Adding an abstract reconcile_users breaks external subclasses.

PasarGuardNode is exported from PasarGuardNodeBridge/__init__.py. Any third-party subclass that does not define reconcile_users now fails to instantiate with TypeError. The other changes in this file are additive keyword parameters and stay compatible.

If backward compatibility matters for this release, provide a default implementation that raises NodeAPIError(501, ...) instead of marking it abstract, and document the new method in the release notes.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@PasarGuardNodeBridge/abstract_node.py` around lines 70 - 77, The abstract
reconcile_users method on PasarGuardNode breaks instantiation of existing
external subclasses. Remove the `@abstractmethod` requirement and provide a
default implementation that raises NodeAPIError with HTTP status 501, preserving
the shown signature; document the new method in the release notes.

tests/test_epoch_fencing.py (1)

45-76: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Bind the loop variables in the closures.

response_for and grpc_request read captured and info_method from the enclosing loop scope. Both are awaited inside the same iteration, so the current test passes. Ruff still reports B023 for both closures, and the sibling tests in tests/test_user_revocation.py already use the default-argument idiom (for example async def transport(_captured=captured, **kwargs)).

Bind the values as default arguments for consistency and to clear the lint finding.

♻️ Proposed change

-                async def response_for(request):
+                async def response_for(request, _captured=captured):
                     if request is None:
                         return service.BaseInfoResponse(
                             started=False,
                             user_sync_epoch_supported=True,
                             user_sync_epoch=40,
                         )
-                    captured.append(request.user_sync_epoch)
+                    _captured.append(request.user_sync_epoch)
@@
-                    async def grpc_request(**kwargs):
-                        request = None if kwargs["method"] is info_method else kwargs["request"]
+                    async def grpc_request(_info_method=info_method, **kwargs):
+                        request = None if kwargs["method"] is _info_method else kwargs["request"]
                         return await response_for(request)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_epoch_fencing.py` around lines 45 - 76, Update the response_for
and grpc_request closures in the node-type loop to bind captured loop values
through default arguments, including captured for response_for and info_method
for grpc_request. Preserve their existing request handling and responses while
eliminating the B023 late-binding lint findings.

Source: Linters/SAST tools

tests/test_user_revocation.py (1)

23-47: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Set _user_sync_epoch_capability_probed explicitly in _controller.

_controller sets _user_sync_epoch_supported = True but leaves _user_sync_epoch_capability_probed unset. The tests pass only because _probe_user_sync_epoch_capability reads the flag with getattr(..., False) and then finds no info attribute on a bare Controller, so it marks the probe complete and returns.

The tests therefore depend on defensive fallbacks rather than on declared state. tests/test_epoch_fencing.py already sets both flags in _configured_node.

♻️ Proposed change

     controller._user_sync_epoch_supported = True
+    controller._user_sync_epoch_capability_probed = True
+    controller._user_sync_epoch_handshake_lock = asyncio.Lock()
+    controller._user_sync_connection_generation = 0

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_user_revocation.py` around lines 23 - 47, Update the test helper
_controller to initialize _user_sync_epoch_capability_probed explicitly
alongside _user_sync_epoch_supported, using the same intended initial state as
_configured_node in tests/test_epoch_fencing.py. Keep the helper’s other
controller state unchanged.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes